package com.qiaofang.housewarehouse.service.impl;

import com.qiaofang.housewarehouse.dao.EsDAO;
import com.qiaofang.housewarehouse.dao.HbaseDao;
import com.qiaofang.housewarehouse.dao.domain.*;
import com.qiaofang.housewarehouse.dao.utils.DateUtils;
import com.qiaofang.housewarehouse.service.HouseCollectionQualityService;
import com.qiaofang.housewarehouse.service.compose.HouseCollectionQualityCompose;
import com.qiaofang.housewarehouse.service.compose.HswhRentalHouseCompose;
import com.qiaofang.housewarehouse.service.compose.HswhSaleHouseCompose;
import com.qiaofang.housewarehouse.stub.bean.HouseCollectionQualityDTO;
import com.qiaofang.housewarehouse.stub.bean.HouseCollectionQualitySourceDTO;
import com.qiaofang.housewarehouse.stub.constants.TableConstants;
import com.qiaofang.housewarehouse.stub.request.HouseCollectionQualityRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Service
public class HouseCollectionQualityServiceImpl implements HouseCollectionQualityService {
    private static final Logger logger = LoggerFactory.getLogger(HouseCollectionQualityServiceImpl.class);

    @Resource
    private EsDAO esDAO;
    @Resource
    private HbaseDao hbaseDao;

    @Override
    public HouseCollectionQualityDTO getHouseCollectionQuality(HouseCollectionQualityRequest houseCollectionQualityRequest) throws Exception {
        HouseCollectionQualityDTO houseCollectionQualityDTO = new HouseCollectionQualityDTO();
        //信息完整度
        String startRowInfo = houseCollectionQualityRequest.getCity() + houseCollectionQualityRequest.getHousingType() + houseCollectionQualityRequest.getHousesBelonging() + houseCollectionQualityRequest.getDataType() + houseCollectionQualityRequest.getStartDate() ;
        String endRowInfo = houseCollectionQualityRequest.getCity() + houseCollectionQualityRequest.getHousingType() + houseCollectionQualityRequest.getHousesBelonging() + houseCollectionQualityRequest.getDataType() + DateUtils.dayAddNum(houseCollectionQualityRequest.getEndDate(), 1);
        List<HswhInfoIntegrity> hswhInfoIntegrities = hbaseDao.listObjectByStartEndRow(TableConstants.HSWH_INFO_INTEGRITY_COUNTERS, startRowInfo, endRowInfo, HswhInfoIntegrity.class, TableConstants.COLUMN_TYPE_LONG);
        houseCollectionQualityDTO.setInfoIntegrity(HouseCollectionQualityCompose.composeDeal(hswhInfoIntegrities));

        String startRow = startRowInfo+"10000000";
        String endRow = houseCollectionQualityRequest.getCity() + houseCollectionQualityRequest.getHousingType() + houseCollectionQualityRequest.getHousesBelonging() + houseCollectionQualityRequest.getDataType()  + houseCollectionQualityRequest.getEndDate() + "90000000";

        List<HswhBrokerHouse> hswhBrokerHouses = hbaseDao.listObjectByStartEndRow(TableConstants.HSWH_BROKER_HOUSE, startRow, endRow,HswhBrokerHouse.class,  TableConstants.COLUMN_TYPE_STRING);
        if (null != hswhBrokerHouses) {
            houseCollectionQualityDTO.setBrokerHouseCount(String.valueOf(hswhBrokerHouses.size()));
            logger.info("查询经纪人{}条", hswhBrokerHouses.size());
        }

        List<HswhRepeatHouse> hswhRepeatHouses = hbaseDao.listObjectByStartEndRow(TableConstants.HSWH_REPEAT_HOUSE,startRow, endRow, HswhRepeatHouse.class, TableConstants.COLUMN_TYPE_STRING);
        if (null != hswhRepeatHouses) {
            houseCollectionQualityDTO.setRepeatHouseCount(String.valueOf(hswhRepeatHouses.size()));
            logger.info("查询重复房源{}条", hswhRepeatHouses.size());
        }

        List<HswhDistinguishTruth> hswhDistinguishTruths = hbaseDao.listObjectByStartEndRow(TableConstants.HSWH_DISTINGUISH_TRUTH,startRow, endRow, HswhDistinguishTruth.class, TableConstants.COLUMN_TYPE_STRING);
        if (null != hswhDistinguishTruths) {
            long countTruth = 0L;
            for(HswhDistinguishTruth hswhDistinguishTruth : hswhDistinguishTruths){
                if(hswhDistinguishTruth.getIsTruth().equals("1")){
                    countTruth = countTruth + 1;
                }
            }
            houseCollectionQualityDTO.setDistinguishTruthCount(String.valueOf(countTruth));
            logger.info("查询真假房源{}条", hswhDistinguishTruths.size());
        }
        List<HswhLimitTimeless> hswhLimitTimelesses = hbaseDao.listObjectByStartEndRow(TableConstants.HSWH_LIMIT_TIMELESS, startRow, endRow, HswhLimitTimeless.class, TableConstants.COLUMN_TYPE_STRING);
        if (null != hswhLimitTimelesses) {
            long countLimit = 0L;
            for(HswhLimitTimeless hswhLimitTimeless : hswhLimitTimelesses){
                if(hswhLimitTimeless.getIsPass().equals("有效")){
                    countLimit = countLimit + 1;
                }
            }
            houseCollectionQualityDTO.setLimitTimelessCount(String.valueOf(countLimit));
            logger.info("查询时效{}条", hswhLimitTimelesses.size());
        }
        return houseCollectionQualityDTO;
    }

    @Override
    public HouseCollectionQualitySourceDTO listHouseCollectionQualitySource(Map<String, String> paramMap) throws Exception {
        HouseCollectionQualitySourceDTO houseCollectionQualitySourceDTO = new HouseCollectionQualitySourceDTO();
        SearchResponse searchResponse = esDAO.searchHouseNoPageList(paramMap);
        logger.info("查询eslticsearch success");
        if (searchResponse != null) {
            SearchHits hits = searchResponse.getHits();
            SearchHit[] hits1 = hits.getHits();
            List<String> rowkeys = new ArrayList<>();
            for (SearchHit hit : hits1) {
                Map<String, Object> source = hit.getSource();
                for (Map.Entry<String, Object> filed : source.entrySet()) {
                    if(filed.getKey().equals("rowkey")){
                        rowkeys.add(filed.getValue().toString());
                    }
                }
            }
            logger.info("从es中获取的二级索引：{}", rowkeys.toString());
            houseCollectionQualitySourceDTO.setTotalCount(String.valueOf(rowkeys.size()));
            if(paramMap.get("indexName").equals(TableConstants.INDEX_RENTAL_HOUSE)){
                houseCollectionQualitySourceDTO.setHswhRentalHouseDetailDTOList(HswhRentalHouseCompose.composeDetailListDTO(hbaseDao.listObjByRowKeys(TableConstants.HSWH_RENTAL_HOUSE, rowkeys, HswhRentalHouse.class)));
            }else{
                houseCollectionQualitySourceDTO.setHswhSaleHouseDetailDTOList(HswhSaleHouseCompose.composeDetailListDTO(hbaseDao.listObjByRowKeys(TableConstants.HSWH_SALE_HOUSE, rowkeys, HswhSaleHouse.class)));
            }
        }
        return houseCollectionQualitySourceDTO;
    }

}
